package com.growthserver.controller;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.growthserver.service.AppletService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.UUID;

@Slf4j
@Api(description = "kafka测试")
@RestController
@RequestMapping("/kafkaTest")
public class KafkaTestController {

    private final Logger LOGGER = LoggerFactory.getLogger(KafkaTestController.class);

    @Autowired
   private KafkaTemplate<String,Object> kafkaTemplate;

    @Value("${askTopic:TOPIC_ASK_INVITE}")
    private String askTopic;


   @ApiOperation(value = "向kafka里面写入消息", notes = "")
   @PostMapping("/sendMsgToKafka")
   public void sendMsgToKafka(@RequestBody String message) {
       log.info("sendMsgToKafka的入参为{}", JSON.toJSONString(message));
       LOGGER.info("sendMsgToKafka的入参为{}", JSON.toJSONString(message));
       JSONObject objects = new JSONObject();
       objects.put("userId","12345567");
       objects.put("integralValue","10000");
       objects.put("growthValue","200000");
       objects.put("id",UUID.randomUUID().toString());
       log.info("参数组装:{}",objects);
       ListenableFuture<SendResult<String, Object>> send = kafkaTemplate.send("TOPIC_INTERNAL_MY_CONSUMER", objects.toJSONString());
       log.info("返回参数:{}",JSON.toJSONString(send));
   }
}